package timeandwindow;

/*
*1.概念生成
* 自定义水位线的生成
*
* 生成水位线的时间：Flink支持在任意算子后面生成水位线，一般建议在source算子后面生成，下游算子感知
*
* was有两个方法：
*   TA：时间戳的创建，创建对象
*      ta：一个核心方法，提取时间
*   WG:创建对象
*      两个方法：
*        oE：给每条数据生成水位线，每条数据到达后都会调用一次该方法
*
*        oP:周期性生成水位线，按照指定周期调用
* 2.1演示有序流，每条生成水位线
* 调出参数快捷键：
* 更好看效果：从端口发数据，注意有序流不处理乱序数据
* 2.2演示有序，周期性生成水位线
* 使用onEvent记录最大时间
* 不写0会有歧义
* 默认周期：200ms，可以调整env调autowmi方法
*
* 3.1 乱序周期生成水位线，注意处理延迟数据
*
* 4.如何优化，合并有序和乱序水位线生成
*
*
*
* */

import com.atguigu.pojo.Event;
import org.apache.flink.api.common.eventtime.*;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
import util.SourceUtil;

import javax.annotation.Nullable;

public class Fink01_UserDefineWatermark {

    /**
     * @param args 哈哈又学会了新功能
     */
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env.getConfig().setAutoWatermarkInterval(500);//周期性生成水位线，时间周期


        SingleOutputStreamOperator<Event> ds = env.socketTextStream("hadoop102", 9999)
                .map(
                        line -> {
                            String[] files = line.split(",");
                            return new Event(files[0].trim(), files[1].trim(), Long.valueOf(files[2].trim()));
                        }
                );
        ds.print("INPUT");


        //生成水位线
        SingleOutputStreamOperator<Event> timestampsAndWatermarks = ds.assignTimestampsAndWatermarks(
                //括号只让传一个参，即生成水位线策略，下面都是new对象

                //1.有序生成水位线
                //new OrderWatermarkStrategy()//todo Strategy:策略

                //2.乱序流生成水位线
                //new UnOrderedWatermarkStrategy()

                //3.有序 + 乱序
                new MyWatermarkStrategy(2000L)
        );


        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    /*
     * 1.有序流水线生成
     * */

    public static class OrderWatermarkStrategy implements WatermarkStrategy<Event> {

        //创建类实现水位线生成策略WatermarkStrategy，重写两个方法，一个
        //是抽象方法即WatermarkGenerator，另外一个不是抽象方法即imestampAssigner

        //两个方法，一个创建时间戳分配器，一个创建水位线生成器

        //水位线生成器
        @Override
        public WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
            return new OrderedWatermarkGenerator();//要求返回水位线生成器对象，参数是Event
        }

        //时间戳分配器
        @Override
        public TimestampAssigner<Event> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
            return new OrderedTimestampAssigner();
        }
    }

    /*
     * 有序流时间戳分配器
     *
     * */
    public static class OrderedTimestampAssigner implements TimestampAssigner<Event>{

        @Override
        public long extractTimestamp(Event event, long l) {
            return event.getTs();
        }
    }

    /*
    * 有序流水线生成器
    *
    * */

    public static class OrderedWatermarkGenerator implements WatermarkGenerator<Event>{

        //实现接口里两个抽象方法
        private Long maxTs = Long.MIN_VALUE;


        /**
         * 给每条数据生成水位线，每条数据到达都会调用一次该方法
         * @param event   当前到达的数据
         * @param eventTimestamp  通过TimestamapAssigner提取时间
         * @param output
         */
        @Override
        public void onEvent(Event event, long eventTimestamp, WatermarkOutput output) {
            // 原则: 以已经到达的数据中最大的时间来生成水位线
            // 因为是有序流，可以保证当前到达的数据的时间一定是最大的， 基于当前数据的时间来生成水位线即可。

            //创建水位线对象
            //Watermark watermark = new Watermark(eventTimestamp);
            //发射水位线
            //output.emitWatermark(watermark);
            //System.out.println("onEvent ==> 有序流每条数据生成水位线: " + watermark );


            // 记录最大时间
           maxTs = eventTimestamp; //这里是什么时间？


        }

          /**
         * 周期性生成水位线
         *
         * 默认周期:  200ms  ,  PipelineOptions.AUTO_WATERMARK_INTERVAL = Duration.ofMillis(200)
         *
         * 设置周期:  env.getConfig().setAutoWatermarkInterval(500);
         */
        @Override
        public void onPeriodicEmit(WatermarkOutput output) {
            //原则：以已经到达的数据中最大的时间来生成水位线
            //创建Watermark对象
            Watermark watermark = new Watermark(maxTs);

            //发射水位线
            output.emitWatermark(watermark);

            System.out.println("onPeriodicEmit ==> 有序流周期性生成水位线：" + watermark);


        }


    }
//todo 2.乱序流生成水位线
    private static class UnOrderedWatermarkStrategy implements WatermarkStrategy<Event>{
    @Override
    public WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
        return new UnOrderedWatermarkGenerator();//抽象类
    }

    @Override
    public TimestampAssigner<Event> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
        return new UnOrderedTimestampAssigner();
    }

    /*
    * 2.乱序流水线生成器
    *
    * */

    public  static class UnOrderedWatermarkGenerator implements WatermarkGenerator<Event>{

        private Long delay = 2000L;
        private Long maxTs = Long.MIN_VALUE + delay; //这里为什么？

        /*
        * 给每条数据生成水位线
        *
        * */
        @Override
        public void onEvent(Event event, long l, WatermarkOutput watermarkOutput) {
             // 原则: 以已经到达的数据中最大的时间来生成水位线
            //      如果考虑迟到数据，需要在生成水位线的时候减去延迟的时间

            // 记录最大时间
            maxTs = Math.max(maxTs , event.getTs()); //跟下面是不是等价的？

            //生成水位线
            //Watermark watermark = new Watermark(maxTs);
            //output.emitWatermark( watermark );
            //System.out.println("onEvent ==> 乱序流每条数据生成水位线: " + watermark );

        }

        /**
         * 周期性生成水位线
         * @param watermarkOutput
         */

        @Override
        public void onPeriodicEmit(WatermarkOutput watermarkOutput) {
            //原则：以已经到达的数据中最大的时间来生成水位线
            //如果考虑迟到数据，需要再生成水位线的时候减去延迟的时间
            //生成水位线
            Watermark watermark = new Watermark(maxTs - delay);
            watermarkOutput.emitWatermark(watermark);
            System.out.println("onPeriodicEmit ==> 乱序流周期性生成水位线：" + watermark);

        }
    }

    /**
     * 乱序流时间戳分配器
     */
    private class UnOrderedTimestampAssigner implements TimestampAssigner<Event> {
        @Override
        public long extractTimestamp(Event element, long recordTimestamp) {
            return element.getTs();
        }
    }
}

/*
* 有序加乱序
* */

    private static class MyWatermarkStrategy implements WatermarkStrategy<Event>{

        private Long delay;

        public MyWatermarkStrategy(Long delay){ this.delay = delay;}

        public MyWatermarkStrategy(long delay) {
            this.delay = delay ;
        }

        @Override
        public WatermarkGenerator<Event> createWatermarkGenerator(WatermarkGeneratorSupplier.Context context) {
            return new MyWatermarkGenerator(delay);
        }

        @Override
        public TimestampAssigner<Event> createTimestampAssigner(TimestampAssignerSupplier.Context context) {
            return new MyTimestampAssigner();
        }

        /*
        * 有序 + 乱序水位线生成
        * */


        private static class MyWatermarkGenerator implements WatermarkGenerator<Event> {
            private Long delay;

            private Long maxTs ;


            public MyWatermarkGenerator(Long delay) {
                this.delay = delay;
                maxTs = Long.MIN_VALUE + delay;
            }




            @Override
            public void onEvent(Event event, long eventTimestamp, WatermarkOutput output) {
                //原则：以已经到达的数据中最大的时间来生成水位线
                //如果考虑迟到数据，需要再生成水位线的时候减去延迟的时间

                //记录最大时间
                maxTs = Math.max( maxTs ,eventTimestamp);

            }

            @Override
            public void onPeriodicEmit(WatermarkOutput output) {
                //生成水位线
                Watermark watermark = new Watermark(maxTs - delay);
                output.emitWatermark( watermark );
                System.out.println("onPeriodicEmit ==> 周期性生成水位线：" + watermark);

            }
        }
/*
* 有序 + 乱序时间戳分配器
*
* */
        private static class MyTimestampAssigner implements TimestampAssigner<Event> {
            @Override
            public long extractTimestamp(Event element, long recordTimestamp) {
                return element.getTs();
            }
        }
    }
}
